package com.bs.service;


import com.bs.domain.ServiceInfo;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.geo.GeoPoint;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.geogrid.GeoHashGrid;
import org.elasticsearch.search.aggregations.bucket.geogrid.GeoHashGridBuilder;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramBuilder;
import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
import org.elasticsearch.search.aggregations.bucket.terms.StringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsBuilder;
import org.elasticsearch.search.aggregations.metrics.geobounds.GeoBounds;
import org.elasticsearch.search.aggregations.metrics.geobounds.GeoBoundsBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.stereotype.Service;

import java.util.*;

@Service
public class PaltformDashService {
    @Autowired
    ElasticsearchTemplate elasticsearchTemplate;


    private String startTime;
    private String endTime;
    private Long interval;

    private String active;

    /**
     * 登录统计
     */

    private SearchResponse getLoginToalResponse() {
        String index = "logstash-*-" + active + "-epes-*";
        Client transportClient = elasticsearchTemplate.getClient();
        BoolQueryBuilder builder = QueryBuilders.boolQuery();

        builder.filter(QueryBuilders.boolQuery().must(QueryBuilders.matchPhrasePrefixQuery("key1.raw", "LOGIN_SUCCESS"))
            .must(QueryBuilders.rangeQuery("@timestamp").from(startTime).to(endTime).timeZone("+08:00")));
        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch(index).setTypes("log");

        return searchRequestBuilder
            .setSize(0)
            .setQuery(builder)
            .execute().actionGet();
    }

    public Map<String, Long> loginTotal(String startTime, String endTime, String active) {
        this.startTime = startTime;
        this.endTime = endTime;
        this.active = active;
        Map<String, Long> map = new HashMap<>();
        SearchResponse response = getLoginToalResponse();
        long loginTotal = response.getHits().getTotalHits();
        map.put("total", loginTotal);
        return map;
    }

    /**
     * 服务调用统计
     *
     * @return
     */
    private SearchResponse getCallServiceTotalResponse() {
        String index = "logstash-*-" + active + "-svcsql-*";
        Client transportClient = elasticsearchTemplate.getClient();
        BoolQueryBuilder builder = QueryBuilders.boolQuery();
        builder.must(QueryBuilders.matchPhrasePrefixQuery("logmsg", "4. service end."));
        builder.filter(QueryBuilders.boolQuery().must(QueryBuilders.termQuery("category", "SVCSQL"))
            .must(QueryBuilders.rangeQuery("@timestamp").from(startTime).to(endTime).timeZone("+08:00")));
        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch(index).setTypes("log");
        return searchRequestBuilder
            .setSize(0)
            .setQuery(builder)
            .execute().actionGet();
    }

    public Map<String, Long> callServiceTotal(String startTime, String endTime, String active) {
        this.startTime = startTime;
        this.endTime = endTime;
        this.active = active;
        Map<String, Long> map = new HashMap<>();
        SearchResponse response = getCallServiceTotalResponse();
        long loginTotal = response.getHits().getTotalHits();
        map.put("total", loginTotal);

        return map;
    }


    /**
     * 画面访问统计
     *
     * @return
     */
    private SearchResponse getFrameTotallResponse() {
        String index = "logstash-*-" + active + "-epes-*";
        Client transportClient = elasticsearchTemplate.getClient();
        BoolQueryBuilder builder = QueryBuilders.boolQuery();

        builder.filter(QueryBuilders.boolQuery().must(QueryBuilders.matchPhrasePrefixQuery("key1.raw", "ACCESSFORM"))
            .must(QueryBuilders.rangeQuery("@timestamp").from(startTime).to(endTime).timeZone("+08:00")));
        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch(index).setTypes("log");

        return searchRequestBuilder
            .setSize(0)
            .setQuery(builder)
            .execute().actionGet();
    }

    public Map<String, Long> frameTotal(String startTime, String endTime, String active) {
        this.startTime = startTime;
        this.endTime = endTime;
        this.active = active;
        Map<String, Long> map = new HashMap<>();
        SearchResponse response = getFrameTotallResponse();
        long loginTotal = response.getHits().getTotalHits();
        map.put("total", loginTotal);
        return map;
    }


    /**
     * 异步路由发送统计
     *
     * @return
     */
    private SearchResponse getAsyncRouteSendTotalResponse() {
        String index = "logstash-*-" + active + "-comm-*";
        Client transportClient = elasticsearchTemplate.getClient();
        BoolQueryBuilder builder = QueryBuilders.boolQuery();

        builder.filter(QueryBuilders.boolQuery().must(QueryBuilders.termQuery("category", "COMM"))
            .must(QueryBuilders.termQuery("progress", "SRC->XCOM_WRITE_SEND"))
            .must(QueryBuilders.rangeQuery("@timestamp").from(startTime).to(endTime).timeZone("+08:00")));
        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch(index).setTypes("log");

        return searchRequestBuilder
            .setSize(0)
            .setQuery(builder)
            .execute().actionGet();
    }

    public Map<String, Long> asyncRouteSendTotal(String startTime, String endTime, String active) {
        this.startTime = startTime;
        this.endTime = endTime;
        this.active = active;
        Map<String, Long> map = new HashMap<>();
        SearchResponse response = getAsyncRouteSendTotalResponse();
        long loginTotal = response.getHits().getTotalHits();
        map.put("total", loginTotal);
        return map;
    }


    /**
     * 异步路由接收统计
     *
     * @return
     */
    private SearchResponse getAsyncRouteReceiveTotalResponse() {
        String index = "logstash-*-" + active + "-comm-*";
        Client transportClient = elasticsearchTemplate.getClient();
        BoolQueryBuilder builder = QueryBuilders.boolQuery();

        builder.filter(QueryBuilders.boolQuery().must(QueryBuilders.termQuery("category", "COMM"))
            .must(QueryBuilders.termQuery("uid", "xcom"))
            .must(QueryBuilders.matchPhrasePrefixQuery("logmsg", "Routing-async called-service begin"))
            .mustNot(QueryBuilders.termQuery("uuid", "<NULL-GUID>"))
            .must(QueryBuilders.rangeQuery("@timestamp").from(startTime).to(endTime).timeZone("+08:00")));
        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch(index).setTypes("log");

        return searchRequestBuilder
            .setSize(0)
            .setQuery(builder)
            .execute().actionGet();
    }

    public Map<String, Long> asyncRouteReceiveTotal(String startTime, String endTime, String active) {
        this.startTime = startTime;
        this.endTime = endTime;
        this.active = active;
        Map<String, Long> map = new HashMap<>();
        SearchResponse response = getAsyncRouteReceiveTotalResponse();
        long loginTotal = response.getHits().getTotalHits();
        map.put("total", loginTotal);
        return map;
    }


    /**
     * 批处理调用统计
     *
     * @return
     */
    private SearchResponse getBatchTotalResponse() {
        String index = "logstash-*-" + active + "-svcsql-*";
        Client transportClient = elasticsearchTemplate.getClient();
        BoolQueryBuilder builder = QueryBuilders.boolQuery();

        builder.filter(QueryBuilders.boolQuery().must(QueryBuilders.termQuery("category", "SVCSQL"))
            .must(QueryBuilders.matchPhrasePrefixQuery("logmsg", "2. batch"))
            .must(QueryBuilders.rangeQuery("@timestamp").from(startTime).to(endTime).timeZone("+08:00")));
        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch(index).setTypes("log");

        return searchRequestBuilder
            .setSize(0)
            .setQuery(builder)
            .execute().actionGet();
    }

    public Map<String, Long> batchTotal(String startTime, String endTime, String active) {
        this.startTime = startTime;
        this.endTime = endTime;
        this.active = active;
        Map<String, Long> map = new HashMap<>();
        SearchResponse response = getBatchTotalResponse();
        long loginTotal = response.getHits().getTotalHits();
        map.put("total", loginTotal);
        return map;
    }

    /**
     * 服务调用走势
     *
     * @return
     */
    private SearchResponse getCallServiceLineResponse() {
        String index = "logstash-*-" + active + "-svcsql-*";
        Client transportClient = elasticsearchTemplate.getClient();
        BoolQueryBuilder builder = QueryBuilders.boolQuery();
        builder.must(QueryBuilders.matchPhrasePrefixQuery("logmsg", "4. service end."))
            .mustNot(QueryBuilders.wildcardQuery("svc", "ep?*"))
            .mustNot(QueryBuilders.termQuery("uid", "xcom"));


        builder.filter(
            QueryBuilders.boolQuery().must(QueryBuilders.termQuery("category", "SVCSQL"))
                .must(QueryBuilders.rangeQuery("@timestamp").from(startTime).to(endTime).timeZone("+08:00")));
        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch(index).setTypes("log");
        DateHistogramBuilder timeAgg = AggregationBuilders.dateHistogram("date_histogram").field("@timestamp").interval(interval).minDocCount(1).timeZone("+08:00");
        TermsBuilder partitionAgg = AggregationBuilders.terms("partition").field("partition").size(20);


        return searchRequestBuilder
            .setSize(0)
            .setQuery(builder)
            .addAggregation(partitionAgg.subAggregation(timeAgg))
            .execute().actionGet();
    }

    public List<ServiceInfo> callServiceLine(String startTime, String endTime, long interval, String active) {
        this.startTime = startTime;
        this.endTime = endTime;
        this.active = active;
        this.interval = interval;
        ArrayList<ServiceInfo> list = new ArrayList<>();
        SearchResponse response = getCallServiceLineResponse();
        System.out.println();
       if (response.getHits().totalHits() != 0) {
            StringTerms deployName = (StringTerms) response.getAggregations().getAsMap().get("partition");
            Iterator<Terms.Bucket> iterator = deployName.getBuckets().iterator();
            while (iterator.hasNext()) {
                HashMap<String, Long> hashMap = new HashMap<>();
                Terms.Bucket next = iterator.next();
                String partition = next.getKeyAsString();
                Histogram agg = next.getAggregations().get("date_histogram");
                for (Histogram.Bucket bucket : agg.getBuckets()
                    ) {
                    String time = bucket.getKeyAsString();
                    long count = bucket.getDocCount();
                    hashMap.put(time, count);
                }
                ServiceInfo serviceInfo = new ServiceInfo(partition, hashMap);
                list.add(serviceInfo);
            }
        }
        return list;
    }

    /**
     * 异步路由调用总体走势
     *
     * @return
     */
    private SearchResponse getAsyncRouteLineResponse() {
        String index = "logstash-*-" + active + "-comm-*";
        Client transportClient = elasticsearchTemplate.getClient();
        BoolQueryBuilder builder = QueryBuilders.boolQuery();
        builder.must(QueryBuilders.matchPhrasePrefixQuery("logmsg", "Routing-async"))
            .mustNot(QueryBuilders.matchPhraseQuery("logmsg", "UUID=[<NULL-GUID>]"));
        builder.filter(
            QueryBuilders.boolQuery().must(QueryBuilders.termQuery("category", "COMM"))
                .must(QueryBuilders.boolQuery()
                    .should(QueryBuilders.termQuery("progress_seq", "500"))
                    .should(QueryBuilders.termQuery("progress_seq", "100")))
                .must(QueryBuilders.rangeQuery("@timestamp").from(startTime).to(endTime).timeZone("+08:00")));
        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch(index).setTypes("log");
        DateHistogramBuilder timeAgg = AggregationBuilders.dateHistogram("date_histogram").field("@timestamp").interval(interval).minDocCount(1).timeZone("+08:00");

        TermsBuilder progressAgg = AggregationBuilders.terms("progress").field("progress").size(2);
        TermsBuilder partitionAgg = AggregationBuilders.terms("partition").field("partition").size(20);
        return searchRequestBuilder
            .setSize(0)
            .setQuery(builder)
            .addAggregation(partitionAgg.subAggregation(progressAgg.subAggregation(timeAgg)))
            .execute().actionGet();
    }

    public Map<String, Object> asyncRouteLine(String startTime, String endTime, long interval, String active) {
        this.startTime = startTime;
        this.endTime = endTime;
        this.active = active;
        this.interval = interval;
        Map<String, Object> map = new HashMap<>();
        ArrayList<ServiceInfo> list1 = new ArrayList<>();
        ArrayList<ServiceInfo> list2 = new ArrayList<>();
        ArrayList<String> listPartition = new ArrayList<>();
        ArrayList<String> listTime = new ArrayList<>();
        SearchResponse response = getAsyncRouteLineResponse();
        if (response.getHits().totalHits() != 0) {
            StringTerms partitionTerms = (StringTerms) response.getAggregations().getAsMap().get("partition");
            Iterator<Terms.Bucket> iterator = partitionTerms.getBuckets().iterator();
            while (iterator.hasNext()) {
                Terms.Bucket next = iterator.next();
                String partitionKey = next.getKeyAsString();
                listPartition.add(partitionKey);
                StringTerms progress = next.getAggregations().get("progress");
                List<Terms.Bucket> buckets = progress.getBuckets();
                for (Terms.Bucket bucket : buckets) {
                    Histogram date = (Histogram) bucket.getAggregations().getAsMap().get("date_histogram");
                    List<? extends Histogram.Bucket> dateBuckets = date.getBuckets();
                    if ("SRC->XCOM->DEST".equals(bucket.getKey())) {
                        for (Histogram.Bucket db : dateBuckets) {
                            HashMap<String, Long> hashMap = new HashMap<>();
                            String dateKey = db.getKeyAsString();
                            long count = db.getDocCount();
                            hashMap.put(dateKey, count);
                            ServiceInfo serviceInfo = new ServiceInfo(partitionKey, hashMap);
                            list1.add(serviceInfo);
                            if (String.valueOf(listTime).indexOf(dateKey) == -1) {
                                listTime.add(dateKey);
                            }
                        }
                    } else {
                        for (Histogram.Bucket db : dateBuckets) {
                            HashMap<String, Long> hashMap = new HashMap<>();
                            String dateKey = db.getKeyAsString();
                            long count = db.getDocCount();
                            hashMap.put(dateKey, count);
                            ServiceInfo serviceInfo = new ServiceInfo(partitionKey, hashMap);
                            list2.add(serviceInfo);
                            if (String.valueOf(listTime).indexOf(dateKey) == -1) {
                                listTime.add(dateKey);
                            }
                        }
                    }
                }
            }
            map.put("SRCXCOMDEST", list1);
            map.put("SRC", list2);
            map.put("partition", listPartition);
            map.put("time", listTime);
        }
        return map;
    }

    /**
     * 地图
     *
     * @return
     */
    private SearchResponse getMapServiceResponse() {
        String index = "logstash-*-" + active + "-svcsql-*";
        Client transportClient = elasticsearchTemplate.getClient();
        BoolQueryBuilder builder = QueryBuilders.boolQuery();
        builder.must(QueryBuilders.matchPhrasePrefixQuery("logmsg", "4. service end."));


        builder.filter(
            QueryBuilders.boolQuery().must(QueryBuilders.termQuery("category", "SVCSQL"))
                .must(QueryBuilders.rangeQuery("@timestamp").from(startTime).to(endTime).timeZone("+08:00")));
        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch(index).setTypes("log");

        GeoHashGridBuilder field = AggregationBuilders.geohashGrid("map").field("geoip.location");
        GeoBoundsBuilder aggregation = AggregationBuilders.geoBounds("agg").field("geoip.location").wrapLongitude(true);
        return searchRequestBuilder
            .setSize(0)
            .setQuery(builder)
            .addAggregation(field.subAggregation(aggregation))
            .execute().actionGet();
    }

    public ArrayList<Object> mapService(String startTime, String endTime, String active) {
        this.startTime = startTime;
        this.endTime = endTime;
        this.active = active;
        ArrayList<Object> list = new ArrayList<>();
        SearchResponse response = getMapServiceResponse();
        if (response.getHits().totalHits() != 0) {
            GeoHashGrid geoHashGrid = (GeoHashGrid) response.getAggregations().getAsMap().get("map");
            Iterator<GeoHashGrid.Bucket> iterator = geoHashGrid.getBuckets().iterator();
            while (iterator.hasNext()) {
                Map<String, Object> hashMap = new HashMap<>();
                GeoHashGrid.Bucket next = iterator.next();
                long count = next.getDocCount();
                GeoBounds agg = next.getAggregations().get("agg");
                GeoPoint bottomRight = agg.bottomRight();
                hashMap.put("total", count);
                hashMap.put("position", bottomRight);
                list.add(hashMap);
            }
        }

        return list;
    }

}
